package cn.tecnova.cd

import cn.tecnova.utils.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Dataset, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark.rdd.EsSpark

/**
  * description:
  * 4188621
  **/
object TecnovaInvestment {
  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      //      .setMaster("local[*]")
      .set("es.index.auto.create", "true")
      .set("es.nodes", ConfigHandler.esNodes)
      .set("es.port", ConfigHandler.esPort)
      .set("es.nodes.wan.only", "true")

    val sc = new SparkContext(conf)
    val sqLContext = new SQLContext(sc)

    import sqLContext.implicits._
    val rowDF = sqLContext.read.jdbc("jdbc:mysql://172.17.22.17:3306/tec_investment","qichacha_investment","id",1,4188621,10,ConfigHandler.props2)

    //导入写es相关的包
    import org.elasticsearch.spark.sql._

    val result: Dataset[TecnovaInvestment] = rowDF.map(row => {
      TecnovaInvestment(
        row.getAs[Long]("id"),
        row.getAs[String]("KeyNo"),
        row.getAs[String]("CompanyId"),
        row.getAs[String]("Name"),
        row.getAs[String]("No"),
        row.getAs[String]("BelongOrg"),
        row.getAs[String]("OperName"),
        row.getAs[String]("StartDate"),
        row.getAs[String]("EndDate"),
        row.getAs[String]("Status"),
        row.getAs[String]("Province"),
        row.getAs[String]("UpdatedDate"),
        row.getAs[String]("ShortStatus"),
        row.getAs[String]("CreditCode"),
        row.getAs[String]("RegistCapi"),
        row.getAs[String]("EconKind"),
        row.getAs[String]("Address"),
        row.getAs[String]("IndustryCode"),
        row.getAs[String]("SubIndustryCode"),
        row.getAs[String]("Industry"),
        row.getAs[String]("SubIndustry"),
        row.getAs[String]("Scope"),
        row.getAs[String]("ContactNumber"),
        row.getAs[String]("Email"),
        row.getAs[String]("WebSite"),
        row.getAs[String]("ImageUrl"),
        row.getAs[String]("OrgNo"),
        row.getAs[String]("EnglishName"),
        row.getAs[String]("Product"),
        row.getAs[Int]("Type"),
        row.getAs[String]("OriginalName"),
        row.getAs[String]("HitField"),
        row.getAs[String]("HitValue"),
        row.getAs[String]("mongo_id")

      )
    })
//    result.saveToEs("tecnova_investment2" + "/company_investment")

    EsSpark.saveToEs(result.rdd,"tecnova_investment/company_investment",Map("es.mapping.id" -> "id"))


    sc.stop()

  }

}


case class TecnovaInvestment(

                              var id: Long,
                              var keyNo: String,
                              var companyId: String,
                              var name: String,
                              var no: String,
                              var belongOrg: String,
                              var operName: String,
                              var startDate: String,
                              var endDate: String,
                              var status: String,
                              var province: String,
                              var updatedDate: String,
                              var shortStatus: String,
                              var creditCode: String,
                              var registCapi: String,
                              var econKind: String,
                              var address: String,
                              var industryCode: String,
                              var subindustryCode: String,
                              var industry: String,
                              var subIndustry: String,
                              var scope: String,
                              var contactNumber: String,
                              var email: String,
                              var webSite: String,
                              var imageUrl: String,
                              var orgNo: String,
                              var englishName: String,
                              var product: String,
                              var `type`: Int,
                              var originalName: String,
                              var hitField: String,
                              var hitValue: String,
                              var mongoId: String
                            )
